package com.wuwangfu.state;

import com.wuwangfu.func.BroadcastProcessFunc;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author: jcshen
 * @Date: 2023-03-10
 * @PackageName: com.wuwangfu.state
 * @ClassName: BroadcastState
 * @Description:
 * @Version: 1.0.0
 * <p>
 * https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/datastream/fault-tolerance/broadcast_state/
 * <p>
 *
 */
public class BroadcastState {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /**
         * 广播维度流
         * INSERT,1,浏览
         * INSERT,2,参与
         * UPDATE,3,消费
         *
         * UPDATE,3,退出
         * DELETE,3,删除
         */
        DataStreamSource<String> lines = env.socketTextStream("localhost", 7777);

        SingleOutputStreamOperator<Tuple3<String, String, String>> dicts = lines.map(new MapFunction<String, Tuple3<String, String, String>>() {
            @Override
            public Tuple3<String, String, String> map(String value) throws Exception {
                String[] fields = value.split(",");
                return Tuple3.of(fields[0], fields[1], fields[2]);
            }
        });
        //广播
        MapStateDescriptor<String, String> dicStateDesctor = new MapStateDescriptor<>("dic-state", String.class, String.class);
        //返回一个广播流
        BroadcastStream<Tuple3<String, String, String>> broadcast = dicts.broadcast(dicStateDesctor);

        /**
         * 业务数据
         *  zs,A,1
         *  lis,A,1
         *  lis,A,1
         *  ww,B,2
         *  zl,A,2
         *
         *  tq,C,3
         *  tq,B,3
         */
        DataStreamSource<String> datas = env.socketTextStream("localhost", 9999);
        SingleOutputStreamOperator<Tuple3<String, String, String>> dataStream = datas.map(new MapFunction<String, Tuple3<String, String, String>>() {
            @Override
            public Tuple3<String, String, String> map(String value) throws Exception {
                String[] fields = value.split(",");

                return Tuple3.of(fields[0], fields[1], fields[2]);
            }
        });
        //分组后 关联维度 数据会更少
        KeyedStream<Tuple3<String, String, String>, Tuple2<String, String>> keyed = dataStream
                .keyBy(new KeySelector<Tuple3<String, String, String>, Tuple2<String, String>>() {
                    @Override
                    public Tuple2<String, String> getKey(Tuple3<String, String, String> tuple) throws Exception {

                        return Tuple2.of(tuple.f1, tuple.f2);
                    }
                });

        keyed.connect(broadcast)
                .process(new BroadcastProcessFunc(dicStateDesctor))
                .print();

//        System.out.println(System.getProperty("file.encoding"));

        env.execute();
    }
}
